package cn.yx.framework.configuration;

import cn.com.yx.ms.domain.mq.MqCommandType;
import cn.com.yx.ms.service.IMqService;
import cn.yx.framework.log.MyLogger;
import cn.yx.framework.util.JSONUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;

//@ServerEndpoint(value = "/ws/{ip}",configurator = SpringConfigurator.class)
@ServerEndpoint(value = "/wsup/{ip}")
public class MyWebSocketUpdate {
    private static final Logger logger = LoggerFactory.getLogger(MyWebSocketUpdate.class);

    IMqService iMqService;

    @Autowired
    public MyWebSocketUpdate(IMqService iMqService) {
        this.iMqService = iMqService;
    }

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<MyWebSocketUpdate> webSocketSet = new CopyOnWriteArraySet<MyWebSocketUpdate>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    public MyWebSocketUpdate() {

    }

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session) throws Exception {

        MyWebSocketUpdate.onlineCount = webSocketSet.size();

        //判断是否重复连接
        MyWebSocketUpdate removeItem = null;

        for (MyWebSocketUpdate item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").equals(session.getPathParameters().get("ip"))) {
                String message = "存在重复连接[" + session.getPathParameters().get("ip") + "],已被强制踢出";
                System.out.println(message);
                sendInfo(session.getPathParameters().get("ip"), message);
                removeItem = item;
                webSocketSet.remove(removeItem);  //从set中删除
                subOnlineCount();//在线数减1
            }
        }

        //session.getPathParameters().get("ip")
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        System.out.println("有新连接【" + this.session.getPathParameters().get("ip") + "】加入！当前在线人数为" + getOnlineCount());
        try {
            //sendMessage(CommonConstant.CURRENT_WANGING_NUMBER.toString());
            sendMessage("有新链接加入了，当前连接数：" + getOnlineCount());
        } catch (IOException e) {
            System.out.println("IO异常");
        }

    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketSet.contains(this)) {
            webSocketSet.remove(this);  //从set中删除
            subOnlineCount();           //在线数减1
            System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
        }
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // System.out.println("来自客户端的消息:" + message);
        System.out.println("[UPCMD]客户端发送============>" + message);

        JSONObject jsonObject = null;
        String command = "";
        String devsn = "";
        boolean jsonValid = JSONUtils.isJSONValid(message);
        if (jsonValid) {

            // region << 指令 >>
            try {

                //解析content
                jsonObject = JSON.parseObject(message);
                command = jsonObject.get("command").toString();
                devsn = jsonObject.get("deviceid").toString();
                MqCommandType mqCommandType = MqCommandType.valueOf("C" + command);
                MyLogger.log("/command/ws-上行/"
                                + devsn
                                + "/"
                                + mqCommandType.getIndex()
                                + "-"
                                + mqCommandType.getName()
                                + "/",
                        message);


                /**
                 * 业务处理
                 * 此处特别说明：由于websocket的接收脱离springboot框架，因此无法取得spring上下文，所有服务的注入都失败
                 * 因此调用服务的方式调用业务方法
                 */

                // 异步
//                List<BasicNameValuePair> urlParams = new ArrayList<>();
//                urlParams.add(new BasicNameValuePair("message", message));
//                HttpClientService.exeAsyncReq(BaseConstant.getServerConfig().getUrl() + "/websocket", true, urlParams, null, null);

//                CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();// 默认的配置

                RequestConfig requestConfig = RequestConfig.custom()
                        .setSocketTimeout(3000).setConnectTimeout(3000).build();
                CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
                        .setDefaultRequestConfig(requestConfig).build();

                try {
                    httpclient.start();
                    final CountDownLatch latch = new CountDownLatch(1);
                    HttpPost request = new HttpPost(BaseConstant.getServerConfig().getUrl() + "/websocket");

                    List<BasicNameValuePair> urlParams = new ArrayList<>();
                    urlParams.add(new BasicNameValuePair("message", message));
                    String getUrl = EntityUtils
                            .toString(new UrlEncodedFormEntity(urlParams));
                    request.setURI(new URI(request.getURI().toString()
                            + "?" + getUrl));
                    Future<HttpResponse> future = httpclient.execute(request, new FutureCallback<HttpResponse>() {
                        //无论完成还是失败都调用countDown()
                        @Override
                        public void completed(final HttpResponse response) {
                            latch.countDown();
                            System.out.println(request.getRequestLine() + "->"
                                    + response.getStatusLine());
                        }
                        @Override
                        public void failed(final Exception ex) {
                            latch.countDown();
                            System.out.println(request.getRequestLine() + "->" + ex);
                        }
                        @Override
                        public void cancelled() {
                            latch.countDown();
                            System.out.println(request.getRequestLine()
                                    + " cancelled");
                        }
                    });

                    // HttpResponse response = future.get();// 获取结果
                    // System.out.println("Response: " + response.getStatusLine());
                    latch.await();
                    System.out.println("Shutting down");
                } finally {
                    httpclient.close();
                    System.out.println("close");
                }


                // 同步
                // HttpClientUtil.sendPostRequest(BaseConstant.getServerConfig().getUrl() + "/websocket/" + URLEncoder.encode(message, "UTF-8"), "", false);
//                String response = HttpClientUtil.sendPostRequest(BaseConstant.getServerConfig().getUrl() + "/websocket", "message=" + message, true);
                MyLogger.log("/command/ws-上行/"
                                + devsn
                                + "/"
                                + mqCommandType.getIndex()
                                + "-"
                                + mqCommandType.getName()
                                + "/",
                        "http请求已发出");

//                switch (mqCommandType) {
//                    case C2002: //公告主请
//                        //iMqService.receive2002(devsn, content);
//                        // HttpClientUtil.sendPostRequest(BaseConstant.getServerConfig().getUrl() + "/websocket/" + URLEncoder.encode(message,"UTF-8"), "", false);
//                        break;
//                    case C2003: //视频主请
//                        //iMqService.receive2003(devsn, content);
//                        break;
//                    case C3001: //设备状态
//                        //iMqService.receive3001(devsn, content);
//                        break;
//                    case C3003: //心跳
//                        //iMqService.receive3003(devsn, content);
//                        break;
//                    case C4008: //远程升级应答
//                        //iMqService.receive4008(devsn, content);
//                        break;
//                    case C4009: //设备截屏
//                        //iMqService.receive4009(devsn, content);
//                        break;
//                }
//
//                for (MyWebSocket item : webSocketSet) {
//                    if (item.session.getPathParameters().get("ip").contains("page")) {
//                        try {
//                            item.sendMessage(message);
//                        } catch (IOException e) {
//                            e.printStackTrace();
//                        }
//                    }
//                }


            } catch (Exception ex) {
                logger.error(ex.getMessage());
                for (StackTraceElement stack :
                        ex.getStackTrace()) {
                    logger.error(stack.getClassName() + "|" + stack.getFileName() + "|" + stack.getMethodName() + "|" + stack.getLineNumber());
                }
            } finally {
                logger.debug("WS接收数据结束...");
            }

            // endregion
        }
    }

    /**
     * 发生错误时调用
     **/
    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("发生错误");
        // error.printStackTrace();
    }


    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
        //this.session.getAsyncRemote().sendText(message);
    }


    /**
     * 群发自定义消息
     */
    public static void sendWebClientInfo(String message) throws IOException {
        for (MyWebSocketUpdate item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").contains("webpage_")) {
                try {
                    item.sendMessage(message);
                } catch (IOException e) {
                    continue;
                }
            }
        }
    }

    public static void sendInfo(String ip, String message) throws Exception {

        // 下行发送日志记录

        for (MyWebSocketUpdate item : webSocketSet) {
            if (item.session.getPathParameters().get("ip").equals(ip)) {
                item.sendMessage(message);
            }
        }
    }

    public static List<String> getClientIps() throws Exception {
        List<String> ips = new ArrayList<>();
        for (MyWebSocketUpdate item : webSocketSet) {
            ips.add(item.session.getPathParameters().get("ip"));
        }
        return ips;
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        MyWebSocketUpdate.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        MyWebSocketUpdate.onlineCount--;
    }


}
